package processFunction;

import beans.SenSorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Process_KeyedProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SenSorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SenSorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        /**
         * 测试keyedProcessed
         * 先分组然后自定义处理
         */
        dataStream.keyBy("id")
                .process(new MyProcess())
                .print();
        env.execute();
    }

    /**
     * 三个参数分别是
     * key的类型、输入和输出类型
     */
    public static class MyProcess extends KeyedProcessFunction<Tuple, SenSorReading, Integer> {
        //定义一个状态来保存定时服务启动的时间
        private ValueState<Long> tsTimetemp;

        @Override
        public void open(Configuration parameters) throws Exception {
            tsTimetemp = getRuntimeContext().getState(new ValueStateDescriptor<Long>("time", Long.class));
        }

        @Override
        public void processElement(SenSorReading senSorReading, Context context, Collector<Integer> collector) throws Exception {
            collector.collect(senSorReading.getId().length());

            //context
//            context.timestamp();//当前的时间
//            context.getCurrentKey();//当前的key
//            context.output();//分流

            //定时服务
            context.timerService().currentProcessingTime();
            context.timerService().currentWatermark();
            context.timerService().registerProcessingTimeTimer(1000l);//处理事件达到一万毫秒执行注册，是从1970年开始计算
            context.timerService().registerEventTimeTimer((senSorReading.getTimeStamp() + 10) * 1000);//当前事件时间发生后的10秒后
            tsTimetemp.update(context.timerService().currentProcessingTime() + 1000l);
//            context.timerService().deleteEventTimeTimer(tsTimetemp.value());//取消
//            context.timerService().deleteProcessingTimeTimer(tsTimetemp.value());//取消


        }

        /**
         * 当定时服务启动是来执行此方法
         *
         * @param timestamp
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
            System.out.println("触发了");
            ctx.getCurrentKey();
//            ctx.output();
            ctx.timeDomain();
        }

        @Override
        public void close() throws Exception {
            tsTimetemp.clear();
        }
    }
}
